package cn.hy.dataSync.task.runner;


import cn.hy.dataSync.config.AbstractTaskDirtyCollector;
import cn.hy.dataSync.config.TaskConfig;
import cn.hy.dataSync.job.State;
import cn.hy.dataSync.statistics.communication.Communication;
import cn.hy.dataSync.statistics.communication.CommunicationTool;
import org.apache.commons.lang.Validate;


public abstract class AbstractRunner {

    private Communication runnerCommunication;

    private AbstractTaskDirtyCollector taskDirtyCollector;

    private final TaskConfig taskConfig;

    private int taskId;

    public AbstractRunner(TaskConfig taskConfig) {
        this.taskConfig = taskConfig;
        this.taskId = taskConfig.getTaskId();
    }

    public void destroy() {
    }

    public State getRunnerState() {
        return this.runnerCommunication.getState();
    }

    private void mark(State state) {
        this.runnerCommunication.setState(state);
        if (state == State.SUCCEEDED) {
            this.runnerCommunication.increaseCounter(CommunicationTool.STAGE, 1);
        }
    }

    public void markRun() {
        mark(State.RUNNING);
    }

    public void markSuccess() {
        mark(State.SUCCEEDED);
    }

    public void markFail(final Throwable throwable) {
        mark(State.FAILED);
        this.runnerCommunication.setTimestamp(System.currentTimeMillis());
        this.runnerCommunication.setThrowable(throwable);
    }

    public TaskConfig getTaskConfig() {
        return taskConfig;
    }

    public int getTaskId() {
        return taskId;
    }

    public void setTaskId(int taskId) {
        this.taskId = taskId;
    }

    public void setRunnerCommunication(final Communication runnerCommunication) {
        Validate.notNull(runnerCommunication, "Communication不能为空");
        this.runnerCommunication = runnerCommunication;
    }

    public Communication getRunnerCommunication() {
        return runnerCommunication;
    }

    public AbstractTaskDirtyCollector getTaskDirtyCollector() {
        return taskDirtyCollector;
    }

    public void setTaskDirtyCollector(AbstractTaskDirtyCollector taskDirtyCollector) {
        this.taskDirtyCollector = taskDirtyCollector;
    }

    /**
     * 关闭
     */
    public abstract void shutdown();
}
